{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(4)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "Chap13_DATA_DIR = ROOT_DIR + \"mnist\" + os.sep\n",
    "\n",
    "Chap13_DENSE_TRAIN_FILE = \"dense_train.ak\"\n",
    "Chap13_DENSE_TEST_FILE = \"dense_test.ak\"\n",
    "\n",
    "PIPELINE_TF_MODEL = \"pipeline_tf_model.ak\"\n",
    "PIPELINE_PYTORCH_MODEL = \"pipeline_pytorch_model.ak\"\n",
    "PIPELINE_ONNX_MODEL = \"pipeline_onnx_model.ak\"\n",
    "\n",
    "AlinkGlobalConfiguration.setPrintProcessInfo(True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_1_3 \n",
    "\n",
    "print(AlinkGlobalConfiguration.getPluginDir())\n",
    "\n",
    "print(\"Auto Plugin Download : \")\n",
    "\n",
    "print(AlinkGlobalConfiguration.getAutoPluginDownload())\n",
    "\n",
    "downloader = AlinkGlobalConfiguration.getPluginDownloader()\n",
    "\n",
    "print(downloader.listAvailablePlugins())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def softmax(train_set, test_set) :\n",
    "    Pipeline()\\\n",
    "        .add(\\\n",
    "            Softmax()\\\n",
    "                .setVectorCol(\"vec\")\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "        )\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .link(\\\n",
    "            EvalMultiClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "    \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def dnn(train_set, test_set) :\n",
    "    Pipeline()\\\n",
    "         .add(\n",
    "            VectorFunction()\\\n",
    "                .setSelectedCol(\"vec\")\\\n",
    "                .setFuncName(\"Scale\")\\\n",
    "                .setWithVariable(1.0 / 255.0)\n",
    "        )\\\n",
    "       .add(\\\n",
    "            VectorToTensor()\\\n",
    "                .setTensorDataType(\"float\")\\\n",
    "                .setSelectedCol(\"vec\")\\\n",
    "                .setOutputCol(\"tensor\")\\\n",
    "                .setReservedCols([\"label\"])\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            KerasSequentialClassifier()\\\n",
    "                .setTensorCol(\"tensor\")\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .setLayers([\n",
    "                    \"Dense(256, activation='relu')\",\n",
    "                    \"Dense(128, activation='relu')\"\n",
    "                ])\\\n",
    "                .setNumEpochs(50)\\\n",
    "                .setBatchSize(512)\\\n",
    "                .setValidationSplit(0.1)\\\n",
    "                .setSaveBestOnly(True)\\\n",
    "                .setBestMetric(\"sparse_categorical_accuracy\")\\\n",
    "                .setNumWorkers(1)\\\n",
    "                .setNumPSs(0)\\\n",
    "       )\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .link(\\\n",
    "            EvalMultiClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "    \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cnn(train_set, test_set) :\n",
    "    Pipeline()\\\n",
    "        .add(\n",
    "            VectorFunction()\\\n",
    "                .setSelectedCol(\"vec\")\\\n",
    "                .setFuncName(\"Scale\")\\\n",
    "                .setWithVariable(1.0 / 255.0)\n",
    "        )\\\n",
    "        .add(\\\n",
    "            VectorToTensor()\\\n",
    "                .setTensorDataType(\"float\")\\\n",
    "                .setTensorShape([28, 28])\\\n",
    "                .setSelectedCol(\"vec\")\\\n",
    "                .setOutputCol(\"tensor\")\\\n",
    "                .setReservedCols([\"label\"])\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            KerasSequentialClassifier()\\\n",
    "                .setTensorCol(\"tensor\")\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .setLayers([\n",
    "                    \"Reshape((28, 28, 1))\",\n",
    "                    \"Conv2D(32, kernel_size=(3, 3), activation='relu')\",\n",
    "                    \"MaxPooling2D(pool_size=(2, 2))\",\n",
    "                    \"Conv2D(64, kernel_size=(3, 3), activation='relu')\",\n",
    "                    \"MaxPooling2D(pool_size=(2, 2))\",\n",
    "                    \"Flatten()\",\n",
    "                    \"Dropout(0.5)\"\n",
    "                ])\\\n",
    "                .setNumEpochs(20)\\\n",
    "                .setValidationSplit(0.1)\\\n",
    "                .setSaveBestOnly(True)\\\n",
    "                .setBestMetric(\"sparse_categorical_accuracy\")\\\n",
    "                .setNumWorkers(1)\\\n",
    "                .setNumPSs(0)\\\n",
    "        )\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .link(\\\n",
    "            EvalMultiClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "    \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_2\n",
    "sw = Stopwatch()\n",
    "sw.start()\n",
    "\n",
    "train_set = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TRAIN_FILE)\n",
    "test_set = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "\n",
    "softmax(train_set, test_set)\n",
    "\n",
    "dnn(train_set, test_set)\n",
    "\n",
    "cnn(train_set, test_set)\n",
    "\n",
    "sw.stop()\n",
    "print(sw.getElapsedTimeSpan())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Chap16_DATA_DIR = ROOT_DIR + \"wine\" + os.sep\n",
    "Chap16_TRAIN_FILE = \"train.ak\";\n",
    "Chap16_TEST_FILE = \"test.ak\";\n",
    "Chap16_FEATURE_COL_NAMES = [\n",
    "    \"fixedAcidity\", \"volatileAcidity\", \"citricAcid\", \"residualSugar\", \"chlorides\",\n",
    "    \"freeSulfurDioxide\", \"totalSulfurDioxide\", \"density\", \"pH\", \"sulphates\",\n",
    "    \"alcohol\"\n",
    "]\n",
    "\n",
    "def linearReg(train_set, test_set) :\n",
    "    LinearRegression()\\\n",
    "        .setFeatureCols(Chap16_FEATURE_COL_NAMES)\\\n",
    "        .setLabelCol(\"quality\")\\\n",
    "        .setPredictionCol(\"pred\")\\\n",
    "        .enableLazyPrintModelInfo()\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .lazyPrintStatistics()\\\n",
    "        .link(\\\n",
    "            EvalRegressionBatchOp()\\\n",
    "                .setLabelCol(\"quality\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "        \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def dnnReg(train_set, test_set) :\n",
    "    Pipeline()\\\n",
    "        .add(\\\n",
    "            StandardScaler()\\\n",
    "                .setSelectedCols(Chap16_FEATURE_COL_NAMES)\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            VectorAssembler()\\\n",
    "                .setSelectedCols(Chap16_FEATURE_COL_NAMES)\\\n",
    "                .setOutputCol(\"vec\")\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            VectorToTensor()\\\n",
    "                .setSelectedCol(\"vec\")\\\n",
    "                .setOutputCol(\"tensor\")\\\n",
    "                .setReservedCols([\"quality\"])\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            KerasSequentialRegressor()\\\n",
    "                .setTensorCol(\"tensor\")\\\n",
    "                .setLabelCol(\"quality\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .setLayers([\n",
    "                    \"Dense(64, activation='relu')\",\n",
    "                    \"Dense(64, activation='relu')\",\n",
    "                    \"Dense(64, activation='relu')\",\n",
    "                    \"Dense(64, activation='relu')\",\n",
    "                    \"Dense(64, activation='relu')\"\n",
    "                ])\\\n",
    "                .setNumEpochs(20)\\\n",
    "                .setNumWorkers(1)\\\n",
    "                .setNumPSs(0)\\\n",
    "       )\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .lazyPrintStatistics()\\\n",
    "        .link(\\\n",
    "            EvalRegressionBatchOp()\\\n",
    "                .setLabelCol(\"quality\")\\\n",
    "                .setPredictionCol(\"pred\")\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "        \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_3\n",
    "sw = Stopwatch()\n",
    "sw.start()\n",
    "\n",
    "train_set = AkSourceBatchOp().setFilePath(Chap16_DATA_DIR + Chap16_TRAIN_FILE)\n",
    "test_set = AkSourceBatchOp().setFilePath(Chap16_DATA_DIR + Chap16_TEST_FILE)\n",
    "\n",
    "linearReg(train_set, test_set)\n",
    "\n",
    "dnnReg(train_set, test_set)\n",
    "\n",
    "sw.stop()\n",
    "print(sw.getElapsedTimeSpan())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "@udf(input_types=[AlinkDataTypes.TENSOR()], result_type=AlinkDataTypes.INT()) \n",
    "def get_max_index(tensor: np.ndarray):\n",
    "    return tensor.argmax().item()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_4_2\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\\\n",
    "        VectorFunctionBatchOp()\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setFuncName(\"Scale\")\\\n",
    "            .setWithVariable(1.0 / 255.0)\n",
    "    )\\\n",
    "    .link(\\\n",
    "        VectorToTensorBatchOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 28, 28, 1])\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"input_1\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\\\n",
    "        TFSavedModelPredictBatchOp()\\\n",
    "            .setModelPath(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_tf.zip\")\\\n",
    "            .setSelectedCols([\"input_1\"])\\\n",
    "            .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\\\n",
    "    )\\\n",
    "    .lazyPrint(3)\\\n",
    "    .link(\\\n",
    "        UDFBatchOp()\\\n",
    "            .setFunc(get_max_index)\n",
    "            .setSelectedCols([\"output_1\"])\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .lazyPrint(3)\\\n",
    "    .link(\\\n",
    "        EvalMultiClassBatchOp()\\\n",
    "            .setLabelCol(\"label\")\\\n",
    "            .setPredictionCol(\"pred\")\\\n",
    "            .lazyPrintMetrics()\n",
    "    )\n",
    "\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_4_3\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\\\n",
    "        VectorFunctionStreamOp()\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setFuncName(\"Scale\")\\\n",
    "            .setWithVariable(1.0 / 255.0)\n",
    "    )\\\n",
    "    .link(\\\n",
    "        VectorToTensorStreamOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 28, 28, 1])\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"input_1\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\\\n",
    "        TFSavedModelPredictStreamOp()\\\n",
    "            .setModelPath(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_tf.zip\")\\\n",
    "            .setSelectedCols([\"input_1\"])\\\n",
    "            .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\n",
    "    )\\\n",
    "    .link(\\\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"output_1\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_4_4\n",
    "PipelineModel(\\\n",
    "    VectorFunction()\\\n",
    "        .setSelectedCol(\"vec\")\\\n",
    "        .setFuncName(\"Scale\")\\\n",
    "        .setWithVariable(1.0 / 255.0),\\\n",
    "    VectorToTensor()\\\n",
    "        .setTensorDataType(\"float\")\\\n",
    "        .setTensorShape([1, 28, 28, 1])\\\n",
    "        .setSelectedCol(\"vec\")\\\n",
    "        .setOutputCol(\"input_1\")\\\n",
    "        .setReservedCols([\"label\"]),\\\n",
    "    TFSavedModelPredictor()\\\n",
    "        .setModelPath(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_tf.zip\")\\\n",
    "        .setSelectedCols([\"input_1\"])\\\n",
    "        .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\\\n",
    ").save(Chap13_DATA_DIR + PIPELINE_TF_MODEL, True)\n",
    "BatchOperator.execute()\n",
    "\n",
    "PipelineModel\\\n",
    "    .load(Chap13_DATA_DIR + PIPELINE_TF_MODEL)\\\n",
    "    .transform(\\\n",
    "        AkSourceStreamOp()\\\n",
    "            .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "    )\\\n",
    "    .link(\\\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"output_1\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_4_5\n",
    "source = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "\n",
    "print(source.getSchemaStr())\n",
    "\n",
    "df = source.firstN(1).collectToDataframe()\n",
    "\n",
    "row = [df.iat[0,0], df.iat[0,1].item()]\n",
    "\n",
    "localPredictor = LocalPredictor(Chap13_DATA_DIR + PIPELINE_TF_MODEL, \"vec string, label int\")\n",
    "\n",
    "print(localPredictor.getOutputSchemaStr())\n",
    "\n",
    "r = localPredictor.map(row)\n",
    "print(str(r[0]) + \" | \" + str(r[2]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_5_2\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\\\n",
    "        VectorToTensorBatchOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 1, 28, 28])\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\\\n",
    "        TorchModelPredictBatchOp()\\\n",
    "            .setModelPath(\n",
    "                \"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt\")\\\n",
    "            .setSelectedCols([\"tensor\"])\\\n",
    "            .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\n",
    "    )\\\n",
    "    .lazyPrint(3)\\\n",
    "    .link(\\\n",
    "        UDFBatchOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"output_1\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .lazyPrint(3)\\\n",
    "    .link(\\\n",
    "        EvalMultiClassBatchOp()\\\n",
    "            .setLabelCol(\"label\")\\\n",
    "            .setPredictionCol(\"pred\")\\\n",
    "            .lazyPrintMetrics()\n",
    "    )\n",
    "\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_5_3\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\\\n",
    "        VectorToTensorStreamOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 1, 28, 28])\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\\\n",
    "        TorchModelPredictStreamOp()\\\n",
    "            .setModelPath(\n",
    "                \"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt\")\\\n",
    "            .setSelectedCols([\"tensor\"])\\\n",
    "            .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\n",
    "    )\\\n",
    "    .link(\\\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"output_1\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_5_4\n",
    "PipelineModel(\\\n",
    "    VectorToTensor()\\\n",
    "        .setTensorDataType(\"float\")\\\n",
    "        .setTensorShape([1, 1, 28, 28])\\\n",
    "        .setSelectedCol(\"vec\")\\\n",
    "        .setOutputCol(\"tensor\")\\\n",
    "        .setReservedCols([\"label\"]),\n",
    "    TorchModelPredictor()\\\n",
    "        .setModelPath(\n",
    "            \"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/mnist_model_pytorch.pt\")\\\n",
    "        .setSelectedCols([\"tensor\"])\\\n",
    "        .setOutputSchemaStr(\"output_1 FLOAT_TENSOR\")\n",
    ").save(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL, True)\n",
    "BatchOperator.execute()\n",
    "\n",
    "PipelineModel\\\n",
    "    .load(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL)\\\n",
    "    .transform(\\\n",
    "        AkSourceStreamOp()\\\n",
    "            .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "    )\\\n",
    "    .link(\\\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"output_1\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_5_5\n",
    "source = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "\n",
    "print(source.getSchemaStr())\n",
    "\n",
    "df = source.firstN(1).collectToDataframe()\n",
    "\n",
    "row = [df.iat[0,0], df.iat[0,1].item()]\n",
    "\n",
    "localPredictor = LocalPredictor(Chap13_DATA_DIR + PIPELINE_PYTORCH_MODEL, \"vec string, label int\")\n",
    "\n",
    "print(localPredictor.getOutputSchemaStr())\n",
    "\n",
    "r = localPredictor.map(row)\n",
    "print(str(r[0]) + \" | \" + str(r[2]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_6_2\n",
    "import json\n",
    "\n",
    "source = RandomTableSourceBatchOp() \\\n",
    "    .setNumRows(100) \\\n",
    "    .setNumCols(10)\n",
    "\n",
    "colNames = source.getColNames()\n",
    "label = \"label\"\n",
    "\n",
    "userParams = {\n",
    "    'featureCols': json.dumps(colNames),\n",
    "    'labelCol': label,\n",
    "    'batch_size': 16,\n",
    "    'num_epochs': 1\n",
    "}\n",
    "\n",
    "tensorFlowBatchOp = TensorFlowBatchOp() \\\n",
    "    .setUserFiles([\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/tf_dnn_batch.py\"]) \\\n",
    "    .setMainScriptFile(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/tf_dnn_batch.py\") \\\n",
    "    .setUserParams(json.dumps(userParams)) \\\n",
    "    .setOutputSchemaStr(\"model_id long, model_info string\") \\\n",
    "    .setNumWorkers(1) \\\n",
    "    .setNumPSs(0)\n",
    "source = source.select(\"*, case when RAND() > 0.5 then 1. else 0. end as label\") \\\n",
    "    .link(tensorFlowBatchOp) \\\n",
    "    .print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_7_1\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\n",
    "        VectorToTensorBatchOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 1, 28, 28])\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\n",
    "        OnnxModelPredictBatchOp()\\\n",
    "            .setModelPath(\n",
    "                \"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/cnn_mnist_pytorch.onnx\")\\\n",
    "            .setSelectedCols([\"tensor\"])\\\n",
    "            .setInputNames([\"0\"])\\\n",
    "            .setOutputNames([\"21\"])\\\n",
    "            .setOutputSchemaStr(\"probabilities FLOAT_TENSOR\")\n",
    "    )\\\n",
    "    .link(\n",
    "        UDFBatchOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"probabilities\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .lazyPrint(3)\\\n",
    "    .link(\n",
    "        EvalMultiClassBatchOp()\\\n",
    "            .setLabelCol(\"label\")\\\n",
    "            .setPredictionCol(\"pred\")\\\n",
    "            .lazyPrintMetrics()\n",
    "    )\n",
    "\n",
    "BatchOperator.execute()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_7_2\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\\\n",
    "    .link(\n",
    "        VectorToTensorStreamOp()\\\n",
    "            .setTensorDataType(\"float\")\\\n",
    "            .setTensorShape([1, 1, 28, 28])\\\n",
    "            .setSelectedCol(\"vec\")\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "            .setReservedCols([\"label\"])\n",
    "    )\\\n",
    "    .link(\n",
    "        OnnxModelPredictStreamOp()\\\n",
    "            .setModelPath(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/cnn_mnist_pytorch.onnx\")\\\n",
    "            .setSelectedCols([\"tensor\"])\\\n",
    "            .setInputNames([\"0\"])\\\n",
    "            .setOutputNames([\"21\"])\\\n",
    "            .setOutputSchemaStr(\"probabilities FLOAT_TENSOR\")\n",
    "    )\\\n",
    "    .link(\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"probabilities\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_7_3\n",
    "PipelineModel(\n",
    "    VectorToTensor()\\\n",
    "        .setTensorDataType(\"float\")\\\n",
    "        .setTensorShape([1, 1, 28, 28])\\\n",
    "        .setSelectedCol(\"vec\")\\\n",
    "        .setOutputCol(\"tensor\")\\\n",
    "        .setReservedCols([\"label\"]),\n",
    "    OnnxModelPredictor()\\\n",
    "        .setModelPath(\"https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/cnn_mnist_pytorch.onnx\")\\\n",
    "        .setSelectedCols([\"tensor\"])\\\n",
    "        .setInputNames([\"0\"])\\\n",
    "        .setOutputNames([\"21\"])\\\n",
    "        .setOutputSchemaStr(\"probabilities FLOAT_TENSOR\")\n",
    ").save(Chap13_DATA_DIR + PIPELINE_ONNX_MODEL, True)\n",
    "BatchOperator.execute()\n",
    "\n",
    "PipelineModel\\\n",
    "    .load(Chap13_DATA_DIR + PIPELINE_ONNX_MODEL)\\\n",
    "    .transform(\n",
    "        AkSourceStreamOp()\\\n",
    "            .setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "    )\\\n",
    "    .link(\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(get_max_index)\\\n",
    "            .setSelectedCols([\"probabilities\"])\\\n",
    "            .setOutputCol(\"pred\")\n",
    "    )\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_7_4\n",
    "source = AkSourceBatchOp().setFilePath(Chap13_DATA_DIR + Chap13_DENSE_TEST_FILE)\n",
    "\n",
    "print(source.getSchemaStr())\n",
    "\n",
    "df = source.firstN(1).collectToDataframe()\n",
    "\n",
    "row = [df.iat[0,0], df.iat[0,1].item()]\n",
    "\n",
    "localPredictor = LocalPredictor(Chap13_DATA_DIR + PIPELINE_ONNX_MODEL, \"vec string, label int\")\n",
    "\n",
    "print(localPredictor.getOutputSchemaStr())\n",
    "\n",
    "r = localPredictor.map(row)\n",
    "print(str(r[0]) + \" | \" + str(r[2]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
